package com.burges.net.tableAPIAndSQL.table

import org.apache.flink.api.common.state.StateDescriptor.Type
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{TableEnvironment, Types}
import org.apache.flink.table.descriptors.{Csv, FileSystem}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/14 
  * 描述     TableFormat传输不同格式类型不同格式的数据
  */
object TableFormat {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val tableEnvironment = TableEnvironment.getTableEnvironment(environment)
		tableEnvironment
				.connect(new FileSystem().path(""))
		//CSVFormat
        		.withFormat(
			new Csv()
        			.field("field1", Types.STRING)  //根据顺序指定字段名称和类型           必选
        			.field("field2", Types.SQL_TIMESTAMP) //根据顺序指定字段名称和类型     必选
        			.fieldDelimiter(",")            // 指定列切割符，默认使用逗号             可选
        			.lineDelimiter("\n")            // 指定行切割符，默认使用换行符            可选
        			.quoteCharacter('"')            // 指定字符串中的单个字符，默认为空        可选
        			.commentPrefix("#")             // 指定Comment的前，默认为空             可选
        			.ignoreFirstLine()                     // 是否忽略第一行                       可选
        			.ignoreParseErrors()                   // 是否忽略解析错误的数据，默认开启        可选
		)

		/**
		  * JSONFormat
		  */
		tableEnvironment
				.connect(new FileSystem().path(""))
//        		.withFormat(
//			        new Json()
//        				.failOnMissingField(true) //当字段缺失的时候是否解析失败          可选
					//方式一，使用Flink数据类型定义，然后通过mapping映射成JSONSchema
//        				.schema(Type.ROW(...))
					//方式二，通过配置jsonSchema构建JSONFormat
//        				.JsonSchema("{type:'object',proprerties:{id:{type:'number',},name:{type:'string'},rowtime:{type:'string',format:'date-time'}}}")
			        //方式三，直接使用TableSchema信息，转换成JSON结构
//					    .deriveSchema()
//		        )

		/**
		  * ApacheAvroFormat
		  */
		tableEnvironment
				.connect(new FileSystem().path(""))
//        		.withFormat(
//			        new Avro()
			        // 通过AvroSpecificRecordClass来定义
//        			    .recordClass(MyRecord.class)
			        // 通过AvroSchema字符串定义
//        		        .avroSchema("{\"type\":\"record\",\"name\":\"event\",\"field\":[{\"name\":\"id\",\"type\":\"long\"},{\"name\":\"name\",\"type\":\"string\"}]}")
//		        )
	}

}
